package com.demo.kafka.flink.test;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;  
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;  
import org.apache.flink.util.Collector;  
  
public class FlatMapExample {  
    public static void main(String[] args) throws Exception {  
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        collectionExample(env);

        env.execute("FlatMap Example");  
    }

    private static void collectionExample(StreamExecutionEnvironment env) {
        DataStream<String> input = env.fromElements("Flink is awesome", "Apache Flink");
//        DataStream<String> input = env.fromElements("Flink is awesome", "Apache Flink");

        DataStream<String> result = input.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String value, Collector<String> out) throws Exception {
                for (String word : value.split(" ")) {
                    out.collect(word);
                }
            }
        });

        result.print();
    }


}